package org.apache.ocean.name;

import java.net.URL;
import java.util.List;
import java.util.Set;
import java.util.concurrent.atomic.AtomicReference;
import java.util.logging.Logger;

import org.apache.commons.lang.StringUtils;
import org.apache.ocean.HttpParameters;
import org.apache.ocean.OceanXML;
import org.apache.ocean.RefWaitNotNull;
import org.apache.ocean.RestartableTimer;
import org.apache.ocean.TimeoutState;

/**
 * 
 * @author jasonr
 */
// TODO: create column or status that tells the master node to commit only on adding a transactionrecord to the marked nodes for semisync
public class NodeNameClient extends NameClient {
	public static long MAX_REPORT_DURATION = 1000 * 30;
	public static Logger log = Logger.getLogger(NodeNameClient.class.getName());
	public String nodeID;
	public RefWaitNotNull<String> cellIDRef = new RefWaitNotNull<String>();
	public RefWaitNotNull<Node> masterNodeRef = new RefWaitNotNull<Node>();
	public Node myChosenMaster;
	public String status;
	public RestartableTimer pollTimer;
	public NodeEventListener nodeEventListener;
	public AtomicReference<List<Node>> nodesToCommitTransactionTo = new AtomicReference<List<Node>>();

	public NodeNameClient(String nodeID, Set<URL> nameURLs, NodeEventListener nodeEventListener) throws Exception {
		super(nameURLs, nodeEventListener);
		this.nodeID = nodeID;
		this.nodeEventListener = nodeEventListener;
	}

	public void startNodeNameClient() throws Exception {
		pollTimer = new RestartableTimer("namenodepoll", new Poll(), 5 * 1000, 30 * 1000);
	}

	public void shutdown() throws Exception {
		super.shutdown();
		pollTimer.shutdown();
	}

	public String getCellIDNoWait() {
		return cellIDRef.getNoWait();
	}

	public String getCellIDWait() throws InterruptedException {
		return cellIDRef.get();
	}

	public boolean isMaster(TimeoutState timeoutState) throws Exception {
		Node masterNode = masterNodeRef.get(timeoutState);
		return StringUtils.equals(masterNode.id, nodeID);
	}

	public boolean isSlave(TimeoutState timeoutState) throws Exception {
		Node masterNode = masterNodeRef.get(timeoutState);
		return !StringUtils.equals(masterNode.id, nodeID);
	}

	public static abstract class NodeEventListener extends EventListener {
		public Node getReportableNode() {
			return null;
		}

		public void masterNodeChanged() {
		}

		public void noCellID() {
		}
	}

	protected class Poll implements RestartableTimer.TimeoutRunnable {
		public void run(TimeoutState timeoutState) {
			try {
				Node returnedNode = reportNodeHttp(nodeEventListener.getReportableNode(), timeoutState);
				if (returnedNode == null) {
					throw new Exception("returnedNode null");
				}
				if (cellIDRef.getNoWait() == null) {
					// no cell id need to obtain one
					if (StringUtils.isBlank(returnedNode.assignedCellID)) {
						nodeEventListener.noCellID();
					} else {
						cellIDRef.set(returnedNode.assignedCellID);
					}
				} else {
					// have cell id, need to get master node
					Node masterNode = chooseAndGetAgreedMasterNode(timeoutState);
					// log.info("masterNode: "+masterNode);
					if (masterNode == null) {
						log.info("no chosen master");
					}
					Node existingMasterNode = masterNodeRef.getNoWait();
					masterNodeRef.set(masterNode);
					//if (masterNode != null && existingMasterNode != null && !StringUtils.equals(masterNode.id, existingMasterNode.id)) {
					//	nodeEventListener.masterNodeChanged();
					//} else if (existingMasterNode == null && masterNode != null) {
				  nodeEventListener.masterNodeChanged();
					//}
				}
			} catch (Throwable throwable) {
				nodeEventListener.exception(throwable);
			}
		}
	}

	public Node chooseAndGetAgreedMasterNode(TimeoutState timeoutState) throws Exception {
		String cellID = cellIDRef.get(timeoutState);
		OceanXML responseOceanXML = getCellNodesHttp(cellID, timeoutState);
		List<Node> nodes = responseOceanXML.bodyToList(new Node.Transform());
		myChosenMaster = NodeUtil.chooseMaster(nodes, responseOceanXML.timestamp);
		//log.info("myChosenMaster: "+myChosenMaster);
		if (myChosenMaster != null) {
			if (StringUtils.equals(myChosenMaster.id, this.nodeID)) {
			}
			// reportChosenMasterHttp(chosenMaster.id, timeoutState);
			Node masterNode = NodeUtil.obtainAgreedMasterNode(nodes, responseOceanXML.timestamp, log);
			if (masterNode != null && StringUtils.equals(myChosenMaster.id, masterNode.id)) {
				return masterNode;
			}
		}
		return null;
	}

	public OceanXML getCellNodesHttp(String cellID, TimeoutState timeoutState) throws Exception {
		HttpParameters parameters = new HttpParameters("action", "getnodes", "cellid", cellID);
		OceanXML oceanXML = httpRequest(parameters, timeoutState);
		return oceanXML;
		// return oceanXML.bodyToList(new Node.Transform());
	}

	private Node reportNodeHttp(Node node, TimeoutState timeoutState) throws Exception {
		if (node.reportedCellID == null && this.cellIDRef.getNoWait() != null) {
			node.reportedCellID = this.cellIDRef.get(timeoutState);
		}
		if (node.reportedMasterID == null && this.myChosenMaster != null) {
			node.reportedMasterID = this.myChosenMaster.id;
		}
		OceanXML request = new OceanXML("action", "reportnode");
		request.addBody(node.toElement());
		OceanXML response = httpRequest(request, timeoutState);
		return (Node) response.getFirstInBody(new Node.Transform());
	}

	/**
	 * public void reportStatusHttp(String status, TimeoutState timeoutState)
	 * throws Exception { HttpParameters parameters = new HttpParameters("action",
	 * "reportnodestatus", "status", status); httpRequest(parameters,
	 * timeoutState); }
	 * 
	 * private Node reportChosenMasterHttp(String chosenMasterID, TimeoutState
	 * timeoutState) throws Exception { HttpParameters parameters = new
	 * HttpParameters("action", "reportchosenmaster", "masterid", chosenMasterID);
	 * OceanXML response = httpRequest(parameters, timeoutState); return
	 * (Node)response.getFirstInBody(new Node.Transform()); }
	 */
	/**
	 * private Node obtainAgreedMasterNode(List<Node> nodes, TimeoutState
	 * timeoutState, Date serverTimestamp) throws Exception {
	 * //log.info("nodes.size: " + nodes.size()); int uniqueMasterCount =
	 * getUniqueReportedMasterCount(nodes); if (uniqueMasterCount != 1) { //throw
	 * new Exception("uniqueMasterCount: " + uniqueMasterCount);'
	 * log.severe("uniqueMasterCount: "+uniqueMasterCount); return null; } try {
	 * String sameMasterID = verifySlavesHaveSameMaster(nodes, 1000 * 5,
	 * serverTimestamp); Node masterNode = (Node)
	 * ClusterUtil.reflectFindByField("id", sameMasterID, nodes);
	 * //log.info("masterNode: "+masterNode); return masterNode; } catch
	 * (Exception exception) { log.log(Level.SEVERE, "nodes don't have same
	 * master: " + nodes); } return null; }
	 * 
	 * private int getUniqueReportedMasterCount(List<Node> nodes) { Set<String>
	 * set = new HashSet<String>(); for (Node node : nodes)
	 * set.add(node.reportedMasterID); return set.size(); }
	 * 
	 * 
	 * Node chooseMaster(List<Node> incoming, Date serverTimestamp) throws
	 * Exception { // filter nodes that have not reported in a while List<Node>
	 * nodes = filterExpiredNodes(incoming, serverTimestamp); ReflectColumns
	 * reflectColumns = new ReflectColumns();
	 * reflectColumns.add("highestCellTransactionIDDatabase",
	 * SortReflect.field("highestCellTransactionIDDatabase"),
	 * SorterConstants.DESCENDING_ORDER); reflectColumns.add("dateCreated",
	 * SortReflect.field("dateCreated"), SorterConstants.ASCENDING_ORDER); Sorter
	 * sorter = Sorter.createReflect(reflectColumns, nodes,
	 * SorterConstants.NULLS_ARE_LEAST); List<Node> sortedNodes =
	 * sorter.sortReflect(); for (Node node : sortedNodes) { if
	 * (ClusterUtil.isStatusOnline(node.reportedStatus)) { return node; } } return
	 * null; }
	 * 
	 * private List<Node> filterExpiredNodes(List<Node> incoming, Date
	 * serverTimestamp) { List<Node> outgoing = new ArrayList<Node>(); for (Node
	 * incomingNode : incoming) { if (incomingNode.reportDate != null) { long
	 * duration = serverTimestamp.getTime() - incomingNode.reportDate.getTime();
	 * if (duration < MAX_REPORT_DURATION) { outgoing.add(incomingNode); } } }
	 * return outgoing; }
	 * 
	 * private String verifySlavesHaveSameMaster(List<Node> nodes, long
	 * desiredDurationAsMaster, Date serverTimestamp) throws Exception { List
	 * slaveStatusList = ClusterUtil.reflectFindFields("reportedStatus",
	 * ClusterConstants.SLAVE, nodes); List masterStatusList =
	 * ClusterUtil.reflectFindFields("reportedStatus", ClusterConstants.MASTER,
	 * nodes); if (slaveStatusList.size() == 0 && masterStatusList.size() == 1) {
	 * Node masterNode = (Node)ClusterUtil.reflectFindByField("reportedStatus",
	 * ClusterConstants.MASTER, nodes); log.info("no slaves, one master"); return
	 * masterNode.id; } String masterID = null; for (Node node : nodes) { if
	 * (StringUtils.equals("slave", node.reportedStatus)) { if (masterID != null &&
	 * !StringUtils.equals(node.reportedMasterID, masterID)) { throw new
	 * Exception("masters do not match node id: " + node.id); } if
	 * (node.reportedMasterIDSinceDate == null) throw new Exception("node " +
	 * node.id + " reportedMasterIDSinceDate null"); // master for at least 5
	 * seconds if (node.reportedMasterIDSinceDate != null) { long durationAsMaster =
	 * serverTimestamp.getTime() - node.reportedMasterIDSinceDate.getTime(); if
	 * (durationAsMaster > desiredDurationAsMaster) { masterID =
	 * node.reportedMasterID; } } else { } } } return masterID; }
	 */
}
